package com.pass.six;

import java.io.IOException;
import java.util.HashSet;
import java.util.Set;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import com.data.ConnectedEdge;
import com.data.Node;
import com.join.SpatialJoin;

public class PassSixReducer extends Reducer<IntWritable, Text, Text, Text> {
	@Override
	protected void reduce(IntWritable key, Iterable<Text> values,
			Context context) throws IOException, InterruptedException {

		Set<Node> nodes = new HashSet<Node>();
		for (Text value : values) {
			nodes.add(new Node(value.toString()));
		}
		Set<ConnectedEdge> edges = SpatialJoin.spatialJoin(nodes);
		Set<ConnectedEdge> removedEdges = new HashSet<ConnectedEdge>();
		for (ConnectedEdge edge : edges) {
			if (edge.isNode()) {
				removedEdges.add(edge);
			}
		}
		edges.removeAll(removedEdges);
		Set<ConnectedEdge> intersectionEdges = new HashSet<ConnectedEdge>();
		for (ConnectedEdge edge : edges) {
			if (edge.getNode1().isIntersection()
					&& edge.getNode2().isIntersection()) {
				intersectionEdges.add(edge);
			} else
				context.write(new Text("1"), new Text("1"));
		}
		for (int i = 0; i < intersectionEdges.size() / 2; i++) {
			context.write(new Text("1"), new Text("1"));
		}
	}
}
